package com.shujia.mapredue;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class Demo6Filter {

    public static class FilterMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String line = value.toString();

            //取出性别为男的学生
            String gender = line.split(",")[3];

            if ("男".equals(gender)) {
                context.write(value, NullWritable.get());
            }
        }
    }


    public static void main(String[] args) throws Exception {

        Job job = Job.getInstance();

        job.setJobName("filter");

        //将 reduce的数量设置为0 ,就不会由reduce
        //在没有聚合操作的时候可以不需要reduce
        //如果没有设置为0  会产生默认的reduce ,默认reduce不会进行任何数据处理
        job.setNumReduceTasks(0);

        job.setJarByClass(Demo6Filter.class);

        job.setMapperClass(FilterMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);


        FileInputFormat.addInputPath(job, new Path("/data/student"));

        FileOutputFormat.setOutputPath(job, new Path("/data/gender"));


        job.waitForCompletion(true);

    }
}
